package kafka.spring_kafka;

import java.util.Random;

import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageChannel;

public class Producer {
	private static final String CONFIG = "/context.xml";
	private static Random rand = new Random();

	public static void main(String[] args) {
		
		final ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext(CONFIG, Producer.class);
		ctx.start();

		final MessageChannel channel = ctx.getBean("inputToKafka", MessageChannel.class);

		for (int i = 0; i < 10; i++) {
		    //默认
		    //boolean flag = channel.send(MessageBuilder.withPayload("消息" + rand.nextInt()).build());
		    //System.out.println(flag);
		    //指定partition 和 topic
			boolean flag = channel.send(MessageBuilder.withPayload("消息" + rand.nextInt()).setHeader("messageKey", String.valueOf(i)).setHeader("topic", "test").build());
			System.out.println(flag);
		}
		ctx.close();
	}
}
